package com.ds.infra.mq.core.storage;

import com.ds.infra.mq.core.protocol.Message;
import com.google.common.collect.EvictingQueue;

import java.util.concurrent.atomic.AtomicLong;

/**
 * 使用google的EvictingQueue实现
 *
 * @author WeiShaoying
 */
public class MessageQueue {

    private final EvictingQueue<Message> queue;

    private final AtomicLong writeOffset = new AtomicLong(0);
    private final AtomicLong readOffset = new AtomicLong(0);

    public MessageQueue(int capacity) {
        this.queue = EvictingQueue.create(capacity);
    }

    public synchronized void put(Message message) {
        queue.add(message);
        writeOffset.incrementAndGet();
    }

    public synchronized Message take() {
        Message message = queue.poll();
        if (message != null) {
            readOffset.incrementAndGet();
        }
        return message;
    }

    public long getWriteOffset() {
        return writeOffset.get();
    }

    public long getReadOffset() {
        return readOffset.get();
    }

    public int size() {
        return queue.size();
    }
}
